package redis

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"gddgame.cc/galaxy/utils"

	"github.com/gomodule/redigo/redis"
)

type monitor struct {
	join   func(node string)
	leave  func(node string)
	finish func()
	key    string

	clusters *utils.Set
}

type register struct {
	node string
	key  string
}

type Adapt struct {
	redisPool *redis.Pool
	duration  time.Duration
	prefix    string

	conn *redis.PubSubConn

	registers sync.Map
	monitors  sync.Map

	subs     sync.Map
	noticing sync.Map

	connecting bool

	ctx context.Context
}

func NewAdapt(redis *redis.Pool, duration time.Duration, prefix string) *Adapt {
	adapt := &Adapt{
		redisPool: redis,
		duration:  duration,
		prefix:    prefix,

		conn:       nil,
		registers:  sync.Map{},
		monitors:   sync.Map{},
		subs:       sync.Map{},
		noticing:   sync.Map{},
		connecting: false,
	}
	return adapt
}

func (a *Adapt) Close() error {
	if !a.Connecting() {
		return nil
	}
	a.connecting = false
	if a.conn != nil {
		_ = a.conn.Unsubscribe()
		_ = a.conn.PUnsubscribe()
		a.ctx.Done()
	}
	a.conn = nil
	return nil
}

func (a *Adapt) connect() {
	if a.Connecting() {
		return
	}
	a.ctx = context.Background()
	// ctx 在close中done
	r, _ := a.redisPool.GetContext(a.ctx)
	conn := &redis.PubSubConn{Conn: r}
	running := func(conn *redis.PubSubConn) {
		ticker := time.NewTicker(a.duration)
		go func() {
			for _ = range ticker.C {
				if !a.connecting {
					ticker.Stop()
					return
				}
				a.notice()
			}
			ticker.Stop()
		}()
		a.connecting = true
		for {
			switch v := conn.Receive().(type) {
			case redis.Message:
				a.handle(v.Channel, v.Data)
			case redis.Subscription:
				a.onSubscription(v.Kind, v.Channel, v.Count)
			case error:
				if strings.Contains(v.Error(), "connection closed") {
					return
				}
			}
		}
	}
	go running(conn)
	a.conn = conn
}

//
//func (a *Adapt) SetNode(node string) error {
//	if a.connecting {
//		return nil
//	}
//	a.node = node
//	return nil
//}

func (a *Adapt) Register(identity string, key string, node string) error {
	if !a.Connecting() {
		a.connect()
	}
	if _, ok := a.registers.Load(identity); ok {
		return fmt.Errorf("Already Register:%s", identity)
	}
	//fmt.Println("register", identity)
	err := a.conn.Subscribe(a.prefix + key + "#node/" + node)
	if err != nil {
		return err
	}
	r := a.redisPool.Get()
	defer r.Close()
	if err := r.Send("publish", a.prefix+key+"#notice", node); err != nil {
		return err
	}
	a.registers.Store(identity, &register{
		key:  key,
		node: node,
	})

	//fmt.Println("Notice success")

	//fmt.Println(a.node, key, err)

	return nil
}

func (a *Adapt) UnRegister(identity string) error {
	if !a.Connecting() {
		return nil
	}
	r, ok := a.registers.Load(identity)
	if !ok {
		return fmt.Errorf("Not Register:%s", identity)
	}
	rr := r.(*register)
	err := a.conn.Unsubscribe(a.prefix + rr.key + "#node/" + rr.node)
	if err != nil {
		return err
	}
	c := a.redisPool.Get()
	defer c.Close()
	if err := c.Send("publish", a.prefix+rr.key+"#notice", rr.node); err != nil {
		return err
	}
	a.registers.Delete(identity)
	return nil
}

func (a *Adapt) Monitor(identity string, key string, join func(node string), leave func(node string), finish func()) error {
	if !a.Connecting() {
		a.connect()
	}
	if _, ok := a.monitors.Load(identity); ok {
		return fmt.Errorf("Already Monitor:%s", identity)
	}
	//fmt.Println("monitor", identity)
	var sub []string
	s, ok := a.subs.Load(key)
	if !ok {
		sub = []string{identity}
		err := a.conn.Subscribe(a.prefix + key + "#notice")
		if err != nil {
			fmt.Println("adapt redis", err)
			return err
		}
		a.subs.Store(key, sub)
	} else {
		sub = s.([]string)
		sub = append(sub, identity)
		sort.Strings(sub)
		a.subs.Store(key, sub)
	}
	monitor := &monitor{
		join:     join,
		leave:    leave,
		finish:   finish,
		clusters: utils.NewSet(),
	}
	a.monitors.Store(identity, monitor)
	a.onNotice(key, sub)
	return nil
}

func (a *Adapt) UnMonitor(identity string) error {
	if !a.Connecting() {
		return nil
	}
	m, ok := a.monitors.Load(identity)
	if !ok {
		return fmt.Errorf("Not Monitor:%s", identity)
	}
	mm := m.(*monitor)
	s, ok := a.subs.Load(mm.key)
	remove := false
	if ok {
		sub := s.([]string)
		result := sort.SearchStrings(sub, identity)
		if result == 0 && sub[0] != identity {

		} else if result == 0 {
			sub = sub[1:]
		} else if result == len(sub)-1 {
			sub = sub[:result]
		} else {
			sub = append(sub[:result], sub[result+1:]...)
		}
		if len(sub) == 0 {
			remove = true
		} else {
			a.subs.Store(mm.key, sub)
		}
	} else {
		remove = true
	}
	if remove {
		a.subs.Delete(mm.key)
		err := a.conn.Unsubscribe(a.prefix + mm.key + "#notice")
		if err != nil {
			return err
		}
	}
	a.monitors.Delete(identity)
	return nil
}

func (a *Adapt) Connecting() bool {
	return a.connecting
}

func (a *Adapt) notice() {
	r := a.redisPool.Get()
	// 保持心跳
	//a.registers.Range(func(key, value interface{}) bool {
	//	r.Send("publish", key.(string)+"#notice", a.node)
	//	return true
	//})
	//a.monitors.Range(func(key, value interface{}) bool {
	//	a.onNotice(value.(*monitor))
	//	return true
	//})
	a.subs.Range(func(key, value interface{}) bool {
		a.onNotice(key.(string), value.([]string))
		return true
	})
	r.Close()
}

func (a *Adapt) onNotice(key string, sub []string) {
	if _, ok := a.noticing.Load(key); ok {
		return
	}
	a.noticing.Store(key, true)
	r := a.redisPool.Get()

	//fmt.Println("Notice success")
	c, err := r.Do("pubsub", "CHANNELS", a.prefix+key+"#node/*")
	//fmt.Println(a.node, key, err)
	if err != nil {
		for _, identity := range sub {
			m, ok := a.monitors.Load(identity)
			if !ok {
				continue
			}
			monitor := m.(*monitor)
			for _, node := range monitor.clusters.List() {
				monitor.leave(node.(string))
			}
			monitor.clusters.Clear()
			monitor.finish()
		}
	} else {
		tmp := utils.NewSet()
		for _, channel := range c.([]interface{}) {
			channelStr := string(channel.([]byte))
			node := strings.TrimPrefix(channelStr, a.prefix+key+"#node/")
			//if node == a.node {
			//	continue
			//}
			tmp.Add(node)
		}
		for _, identity := range sub {
			m, ok := a.monitors.Load(identity)
			if !ok {
				continue
			}
			monitor := m.(*monitor)
			flag := false
			for _, node := range tmp.List() {
				if !monitor.clusters.Has(node) {
					monitor.join(node.(string))
					flag = true
				}
			}
			for _, node := range monitor.clusters.List() {
				if !tmp.Has(node) {
					flag = true
					monitor.leave(node.(string))
				}
			}
			//fmt.Println(key, tmp.List(), monitor.clusters.List(), flag)
			monitor.clusters = tmp
			if flag {
				monitor.finish()
			}
		}
	}

	a.noticing.Delete(key)
	r.Close()
}

func (a *Adapt) handle(channel string, message []byte) {
	if strings.HasSuffix(channel, "#notice") {
		key := strings.TrimPrefix(strings.TrimSuffix(channel, "#notice"), a.prefix)
		sub, ok := a.subs.Load(key)
		if !ok {
			return
		}
		a.onNotice(key, sub.([]string))
		//node := string(message)
		//if !monitor.clusters.Has(node) {
		//	monitor.join(node)
		//	monitor.clusters.Add(node)
		//}
	} else {

	}
}

func (a *Adapt) onSubscription(kind string, channel string, count int) {
	//switch kind {
	//// Kind is "subscribe", "unsubscribe", "psubscribe" or "punsubscribe"
	//case "subscribe":
	//case "unsubscribe":
	//}
}
